#include "broker.h"
#include "method_thread.h"
#include "broker_protocol.pb.h"
#include "trace.h"

#include <string.h>
#include <sys/types.h>
#include <sys/socket.h>
#include <sys/un.h>
#include <assert.h>
#include <unistd.h>
#include <errno.h>
#include <fcntl.h>
#include <netinet/in.h>
#include <netinet/tcp.h>
#include <limits.h>

#include <cstdio>
#include <sstream>

const int OPEN_MAX = _POSIX_OPEN_MAX;

namespace {
    bool write_all(int fd, const ::google::protobuf::Message &msg)
    {
        bool r = true;
        uint32_t msg_sz = htonl(msg.ByteSize());
        int total_sz = sizeof(uint32_t) + msg.ByteSize();
        char *buf = new char[total_sz];
        memcpy(buf, &msg_sz, sizeof(uint32_t));
        if (!msg.SerializeToArray(buf + sizeof(uint32_t), msg.ByteSize())) {
            TERR("error occurred while serializing data");
            r = false;
        } else {
            int n, solong = 0;
            while (solong < total_sz) {
                n = write(fd, buf + solong, total_sz - solong);
                if (n == -1) {
                    TERR("write message error: %s", strerror(errno));
                    r = false;
                    break;
                }
                solong += n;
            }
        }
        delete [] buf;
        return r;
    }

    void set_nonblock(int fd)
    {
        int flags = fcntl(fd, F_GETFL, NULL);
        flags |= O_NONBLOCK;
        if (fcntl(fd, F_SETFL, flags) != 0) {
            perror("set_nonblock");
        }
    }

}

cvmfs_broker::cvmfs_broker(const std::string &primary, const std::string &id,
        const std::vector<std::string> &options)
    : hermes_(id, options), vmgr_(primary, &hermes_, this)
{
    pthread_mutex_init(&sub_m_, NULL);
    pthread_mutex_init(&out_m_, NULL);
    pthread_cond_init(&out_cv_, NULL);

    usock_fd_ = socket(AF_UNIX, SOCK_STREAM, 0);
    if (usock_fd_ < 0) {
        perror("cvmfs_broker");
        throw std::string("cannot open a local socket");
    }

    set_nonblock(usock_fd_);

    if (pipe(pipe_fd_) != 0) {
        perror("cvmfs_broker pipe");
        throw std::string("failed to open a pipe");
    }
    set_nonblock(pipe_fd_[0]);

    struct sockaddr_un sun;
    memset(&sun, 0, sizeof(sun));
    sun.sun_family = AF_UNIX;
    snprintf(sun.sun_path, sizeof(sun.sun_path), "/tmp/cvmfs_broker-%d",
            getpid());
    sock_name_ = sun.sun_path;
    TINFO("plugin sock bound to %s", sock_name_.c_str());

    if (bind(usock_fd_, (struct sockaddr *)&sun, sizeof(sun)) < 0) {
        perror("cvmfs_broker bind");
    }

    if (listen(usock_fd_, 5) < 0) {
        perror("cvmfs_broker listen");
    }

#ifdef __linux__
    poller_ = new EPollAIO;
#else
    poller_ = new SelectAIO;
#endif
    poller_->watch_fd(usock_fd_, PM_RDONLY);
    poller_->watch_fd(pipe_fd_[0], PM_RDONLY);

    assert((poll_th_ = method_thread(this, false, &cvmfs_broker::poll_loop))
            != 0);
    assert((output_th_ = method_thread(this, false,
                    &cvmfs_broker::output_loop)) != 0);

}

cvmfs_broker::~cvmfs_broker()
{
    int one = 1;
    if (write(pipe_fd_[1], &one, sizeof(one)) != sizeof(one)) {
        perror("cvmfs_broker pipe write");
    }
    
    pthread_join(poll_th_, NULL);

    delete poller_;

    if (unlink(sock_name_.c_str()) != 0) {
        perror("cvmfs_broker unlink");
    }
}

void cvmfs_broker::poll_loop()
{
    std::vector<int> readable, writable;
    while (true) {
        readable.clear();
        writable.clear();
        poller_->wait_ready(readable, writable);
        std::vector<int>::iterator itr;
        for (itr = readable.begin(); itr != readable.end(); ++itr) {
            int fd = *itr;
            if (fd == usock_fd_) {
                process_new_plugin(fd);
            } else if (fd == pipe_fd_[0]) {
                int data;
                int n = read(pipe_fd_[0], &data, sizeof(data));
                if (n == sizeof(data)) {
                    assert(data == 1);
                    goto out;
                }
            } else {
                process_plugin_msg(fd);
            }
        }
        for (itr = writable.begin(); itr != writable.end(); ++itr) {
            // TODO fill this in
        }
    }
out:
    return;
}

void cvmfs_broker::output_loop()
{
    pthread_mutex_lock(&out_m_);
    while (true) {
        while (out_q_.empty()) {
            pthread_cond_wait(&out_cv_, &out_m_);
        }
        DataMessage *msg = out_q_.front();
        out_q_.pop_front();
        hermes_.deliver(msg);
        delete msg;
    }
    pthread_mutex_unlock(&out_m_);
}

void cvmfs_broker::process_new_plugin(int fd)
{
    sockaddr_un addr;
    socklen_t slen;
    slen = sizeof(addr);
    int clt = accept(usock_fd_, (sockaddr *)&addr, &slen);
    // XXX locking
    if (clt > 0) {
        set_nonblock(clt);
        ViewMessage welcome;
        welcome.set_me(hermes_.id());
        unsigned vid;
        std::vector<std::string> cohorts;
        vmgr_.get_view(&vid, welcome.mutable_primary(), &cohorts);
        welcome.set_newvid(vid);
        for (unsigned i = 0; i < cohorts.size(); ++i) {
            welcome.add_cohorts(cohorts[i]);
        }

        write_all(clt, welcome);
        plugin_fds_.insert(clt);
        poller_->watch_fd(clt, PM_RDWR);
    } else {
        TERR("cvmfs_broker failed to accept: %s", strerror(errno));
    }
}

void cvmfs_broker::process_plugin_msg(int fd)
{
    int msg_sz;
    int n = read(fd, &msg_sz, sizeof(msg_sz));
    if (n == sizeof(msg_sz)) {
        msg_sz = ntohl(msg_sz);
        char *buf = new char[msg_sz];
        int solong = 0;
        while ((n = read(fd, buf + solong, msg_sz - solong)) > 0) {
            solong += n;
        }
        // TODO reuse allocated message objects
        MessageWrapper msg;
        if (msg.ParseFromArray(buf, msg_sz)) {
            switch (msg.mtype()) {
            case MessageWrapper::SUB:
                {
                    TDEBUG("plugin %d said it was interested in %s",
                            fd, msg.subject().c_str());
                    pthread_mutex_lock(&sub_m_);
                    std::list<fd_subscriber>::iterator it;
                    it = plugin_subscribers_.insert(plugin_subscribers_.end(),
                            fd_subscriber(fd));
                    hermes_.subscribe(msg.subject(), &(*it));
                    pthread_mutex_unlock(&sub_m_);
                }
                break;
            case MessageWrapper::DATA:
                pthread_mutex_lock(&out_m_);
                out_q_.push_back(msg.release_data());
                pthread_cond_signal(&out_cv_);
                pthread_mutex_unlock(&out_m_);
                break;
            default:
                break;
            }
        } else {
            TERR("protobuf parse error");
        }
    } else {
        if (n < 0 && errno == ECONNRESET) {
            TERR("connection reset by client");
        } else if (n == 0){
            TDEBUG("connection closed by client");
        } else {
            TERR("cvmfs_broker read: %s", strerror(errno));
        }
        poller_->unwatch_fd(fd, PM_RDWR);
        close(fd);
    }
}

void cvmfs_broker::internal_notify(const std::string &subject,
        const std::string &body)
{
    DataMessage msg;
    msg.set_subject(subject);
    msg.set_body(body);
    std::set<int>::iterator it, end;
    for (it = plugin_fds_.begin(), end = plugin_fds_.end(); it != end; ++it) {
        write_all(*it, msg);
    }
}

void cvmfs_broker::previewchange(unsigned vid)
{
    // tell all plugins that we have an ongoing view change, so do not
    // answer view-dependent request
    std::ostringstream oss;
    oss << vid;
    internal_notify("PREVC", oss.str());
}

void cvmfs_broker::postviewchange(unsigned oldvid, unsigned newvid,
        const std::string &primary,
        const std::vector<std::string> &cohorts)
{
    // tell all plugins that we have a finished view change
    ViewMessage vmsg;
    vmsg.set_oldvid(oldvid);
    vmsg.set_newvid(newvid);
    vmsg.set_primary(primary);
    for (unsigned i = 0; i < cohorts.size(); ++i) {
        vmsg.add_cohorts(cohorts[i]);
    }
    TDEBUG("notifying postvc");
    internal_notify("POSTVC", vmsg.SerializeAsString());
}

